package net.bw.realtime.tmall.dws.funcion1;

import net.bw.realtime.tmall.dwd.bean.TransactionEventBean;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import net.bw.realtime.tmall.common.utils.JdbcUtil;

import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @Author: liuyawei
 * @Date: 2025/5/27
 */

//todo d 地区分布统计
public class ToAreaDisWindowFunction {

	public static void toAreaDistribution(DataStream<TransactionEventBean> timeStream) {

		SingleOutputStreamOperator<String> area_distribution = timeStream
				.filter(x -> "pay_success".equals(x.getActionType()))
				.keyBy(x -> x.getProvince()+ "," + x.getCity())
				.window(TumblingEventTimeWindows.of(Time.seconds(10)))
				.aggregate(new AggregateFunction<TransactionEventBean, Tuple4<Double, Long, String, String>, Tuple4<Double, Long, String, String>>() {
					@Override
					public Tuple4<Double, Long,String, String> createAccumulator() {
						return Tuple4.of(0.0, 0L, "", "");
					}

					@Override
					public Tuple4<Double, Long,String, String> add(TransactionEventBean TransactionEventBean, Tuple4<Double, Long, String, String> a1) {
						Double amount = TransactionEventBean.getQuantity() * TransactionEventBean.getPrice();
						return Tuple4.of(a1.f0 + amount,
								a1.f1 + 1,
								TransactionEventBean.getProvince(),
								TransactionEventBean.getCity()
						);
					}

					@Override
					public Tuple4<Double, Long,String, String> getResult(Tuple4<Double, Long,String, String> a1) {
						return a1;
					}

					@Override
					public Tuple4<Double, Long,String, String> merge(Tuple4<Double, Long,String, String> a1, Tuple4<Double, Long,String, String> acc1) {
						return Tuple4.of(a1.f0 + acc1.f0,a1.f1 + acc1.f1, acc1.f2, acc1.f3);
					}
				}, new ProcessWindowFunction<Tuple4<Double, Long,String, String>, String, String, TimeWindow>() {
					private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");

					@Override
					public void process(String s, ProcessWindowFunction<Tuple4<Double, Long,String, String>, String, String, TimeWindow>.Context context, Iterable<Tuple4<Double, Long,String, String>> iterable, Collector<String> collector) throws Exception {
						Tuple4<Double, Long,String, String> next = iterable.iterator().next();
						String window_start = sdf.format(new Date(context.window().getStart()));
						String window_end = sdf.format(new Date(context.window().getEnd()));

						collector.collect(window_start + "," + window_end + "," + next.f2  + "," + next.f3 + "," + next.f1 + "," + next.f0 + "," + System.currentTimeMillis());
					}
				})
				.name("area-distribution-aggregation")
				.uid("area-distribution-aggregation")
				;


		//area_distribution.print();

		JdbcUtil.sinkToClickhouseUpsert(area_distribution,
				"INSERT INTO jtp_tmall_report.dws_area_daily_report (\n" +
						"    window_start_time, window_end_time, province, city, order_count, total_amount, ts\n" +
						") VALUES (\n" +
						"    ?, ?, ?, ?, ?, ?, ?\n" +
						")"
		);
		;

	}


}
